package com.juma.cartridge.mqbox.receive;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;

import com.juma.cartridge.mqbox.HessianMessageConverter;
import com.juma.cartridge.mqbox.MQContentType;

/**
 * 基于spring-amqp的消息监听器基类
 * @ClassName MessageBoxListener
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年11月13日 下午3:36:44
 * @Version 1.0
 *
 */
@Slf4j
public abstract class MessageBoxListener implements MessageListener {

    private Map<MQContentType,MessageConverter> supportConverters;
    
    private List<MessageReceiveProcessor> messageReceiveProcessors;
    
    public MessageBoxListener() {
        supportConverters = new HashMap<MQContentType,MessageConverter>();
        supportConverters.put(MQContentType.json,new Jackson2JsonMessageConverter());
        supportConverters.put(MQContentType.hessian,new HessianMessageConverter());
        messageReceiveProcessors = new LinkedList<MessageReceiveProcessor>();
    }

    
    public final void onMessage(Message message) {
        message = processMessage(message);
        String messageId = message.getMessageProperties().getMessageId();
        try {
            Object object = convertToObject(message);
            onReceived(messageId,object);        
        }catch(Exception e) {
            log.error("消息处理失败[messageId={"+messageId+"}]::"+e.getMessage(),e);   
            throw e;
        }
    }
    
    private Message processMessage(Message message) {
        for (MessageReceiveProcessor processor:messageReceiveProcessors) {
            message = processor.receiveProcessMessage(message);
        }
        return message;
    }

    
    protected Object convertToObject(Message message) {
        String ctype = message.getMessageProperties().getContentType();
        MQContentType contentType = MQContentType.valueOf(ctype);
        MessageConverter converter = supportConverters.get(contentType);
        return converter.fromMessage(message);
    }


    
    /**
     * 处理接收到的消息
     * @param messageId 消息唯一ID
     * @param object 消息体对象
     */
    protected abstract void onReceived(String messageId,Object object);


    public void setMessageReceiveProcessors(List<MessageReceiveProcessor> messageReceiveProcessors) {
        this.messageReceiveProcessors.addAll(messageReceiveProcessors);
    }
    
}
